package service

import (
	"com.go-micro.code/todolist/go-micro-todolist/mqServer/model"
	"encoding/json"
	"github.com/Shopify/sarama"
	"github.com/spf13/viper"
	"log"
)

func CreateTask() {
	topic := viper.GetString("kafka.topic")
	partionList, err := model.MQ.Partitions(topic) //根据topic找到所有哦分区
	if err != nil {
		log.Fatalln(err)
	}
	//遍历所有分区
	for partion := range partionList {
		pc, err := model.MQ.ConsumePartition(topic, int32(partion), sarama.OffsetNewest)
		if err != nil {
			log.Fatalln(err)
		}
		defer pc.AsyncClose()
		//异步消费
		go func(consumer sarama.PartitionConsumer) {
			for msg := range pc.Messages() {
				var tsk model.Task
				err = json.Unmarshal(msg.Value, &tsk)
				if err != nil {
					panic(err)
				}
				//fmt.Printf("task: %#v\n", tsk)
				err = model.DB.Create(&tsk).Error
				if err != nil {
					panic(err)
				}
			}
		}(pc)
	}
	//阻塞 持续接收消息
	select {}
}
